package com.wy.sloth.redis.service;

import static com.wy.sloth.constant.SlothConstants.REDIS_NAMESPACE_SEPARATOR;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.SortParameters.Order;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SetOperations;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.ZSetOperations.TypedTuple;
import org.springframework.data.redis.core.query.SortQueryBuilder;

import com.alibaba.fastjson.JSONObject;
import com.wy.sloth.config.SlothConfig;
import com.wy.sloth.constant.SlothConstants;
import com.wy.sloth.constant.SlothOrder;
import com.wy.sloth.core.SlothNameManager;
import com.wy.sloth.core.SlothStandardNameManager;
import com.wy.sloth.exception.SlothException;
import com.wy.sloth.model.SlothMessage;
import com.wy.sloth.model.SlothPage;
import com.wy.sloth.redis.lock.LockCallbackHandler;
import com.wy.sloth.redis.lock.RedisLockTemplate;
import com.wy.sloth.util.CollectionUtils;
import com.wy.sloth.util.StringUtils;

/**
 * redis相关服务
 * 
 * @author wangyuan
 * @since 2019年2月15日
 */
public class SlothRedisService {
    
    private static final Logger logger = LoggerFactory.getLogger(SlothRedisService.class);

    private ZSetOperations<String, String> zSetOperations;
    
    private HashOperations<String, String, JSONObject> hashOperations;
    
    private HashOperations<String, String, Object> objectHashOperations;
    
    private SetOperations<String, String> setOperations;
    
    private ValueOperations<String, String> valueOperations;
    
    private SlothNameManager slothNameManager;
    
    private RedisLockTemplate redisLockTemplate;
    
    public SlothRedisService(RedisTemplate<String, JSONObject> redisTemplate, 
            RedisTemplate<String, String> stringRedisTemplate, 
            SlothNameManager slothNameManager, 
            RedisLockTemplate redisLockTemplate) {
        super();
        this.valueOperations = stringRedisTemplate.opsForValue();
        this.zSetOperations = stringRedisTemplate.opsForZSet();
        this.setOperations = stringRedisTemplate.opsForSet();
        this.hashOperations = redisTemplate.opsForHash();
        this.objectHashOperations = redisTemplate.opsForHash();
        
        this.slothNameManager = slothNameManager;
        this.redisLockTemplate = redisLockTemplate;
    }
    
    public SlothRedisService(RedisTemplate<String, JSONObject> redisTemplate, 
            RedisTemplate<String, String> stringRedisTemplate, SlothConfig slothConfig) {
        super();
        this.valueOperations = stringRedisTemplate.opsForValue();
        this.zSetOperations = stringRedisTemplate.opsForZSet();
        this.setOperations = stringRedisTemplate.opsForSet();
        this.hashOperations = redisTemplate.opsForHash();
        this.objectHashOperations = redisTemplate.opsForHash();
        
        this.slothNameManager = new SlothStandardNameManager(slothConfig);
        this.redisLockTemplate = new RedisLockTemplate(slothConfig, stringRedisTemplate);
    }

    /**
     * 获取消息type对应的可处理队列名称集合
     * @param type
     * @return
     */
    public List<String> listQueueNameByType(String type) {
        int score = Integer.parseInt(type);
        
        Set<String> values = zSetOperations.rangeByScore(slothNameManager.getType2QueueKey(), score, score);
        if(CollectionUtils.isEmpty(values)){
            return null;
        }
        
        //计算value固定前缀以及长度
        String valuePrefix = type + REDIS_NAMESPACE_SEPARATOR;
        //截取队列名称
        List<String> queueNames = new ArrayList<>();
        for(String item : values){
            queueNames.add(item.substring(valuePrefix.length()));
        }
        
        return queueNames;
    }
    
    /**
     * 获取队列对应的消息type集合
     * @return
     */
    public Map<String, List<String>> listQueue2Type() {
    	Map<String, List<String>> queue2Types = new HashMap<>();
    	
    	Set<TypedTuple<String>> tuples = zSetOperations.rangeWithScores(slothNameManager.getType2QueueKey(), 0, Long.MAX_VALUE);
    	if(CollectionUtils.isEmpty(tuples)) {
    		return queue2Types;
    	}
    	
        //截取队列名称
    	List<String> types = null;
    	String queueName = null;
        for(TypedTuple<String> item : tuples){
        	queueName = item.getValue().substring((item.getScore().intValue() + REDIS_NAMESPACE_SEPARATOR).length());
        	
        	types = queue2Types.get(queueName);
        	if(types == null) {
        		types = new ArrayList<>();
        		queue2Types.put(queueName, types);
        	}
        	types.add(String.valueOf(item.getScore().intValue()));
        }
        
    	return queue2Types;
    }
    
    /**
     * 获取处理消息的队列名称集合
     * @return
     */
    public List<String> listQueueName() {
    	List<String> queueNames = new ArrayList<>();
    	
    	Set<TypedTuple<String>> tuples = zSetOperations.rangeWithScores(slothNameManager.getType2QueueKey(), 0, Long.MAX_VALUE);
    	if(CollectionUtils.isEmpty(tuples)) {
    		return queueNames;
    	}
    	
        //截取队列名称
    	Set<String> queueNameSet = new HashSet<>();
        for(TypedTuple<String> item : tuples){
        	queueNameSet.add(item.getValue().substring((item.getScore().intValue() + REDIS_NAMESPACE_SEPARATOR).length()));
        }
    	
        queueNames.addAll(queueNameSet);
        
    	return queueNames;
    }
    
    /**
     * 清理消息队列
     * @param queueName
     */
    public void clearMsgQueue(String queueName) {
    	Set<TypedTuple<String>> tuples = zSetOperations.rangeWithScores(slothNameManager.getType2QueueKey(), 0, Long.MAX_VALUE);
    	if(CollectionUtils.isEmpty(tuples)) {
    		return;
    	}
    	
    	List<String> values = new ArrayList<>();
    	String tempQueueName = null;
        for(TypedTuple<String> item : tuples){
        	tempQueueName = item.getValue().substring((item.getScore().intValue() + REDIS_NAMESPACE_SEPARATOR).length());
        	
        	if(tempQueueName.equals(queueName)) {
        		values.add(item.getValue());
        	}
        }
        
        for(String valueItem : values) {
        	zSetOperations.remove(slothNameManager.getType2QueueKey(), valueItem);
        }
    }
    
    /**
     * 注册消息type与可处理队列关系到redis
     * @param type
     * @param queueName
     * @return
     */
    public Boolean registerTypeAndQueueNameRelation(String type, String queueName) {
        return zSetOperations.add(slothNameManager.getType2QueueKey(), 
        		type + REDIS_NAMESPACE_SEPARATOR + queueName, Integer.parseInt(type));
    }
    
    /**
     * 从redis中删除消息type与可处理队列关系
     * @param type
     * @param queueName
     * @return
     */
    public Long deleteTypeAndQueueNameRelation(final String type, final String queueName) {
        return zSetOperations.remove(slothNameManager.getType2QueueKey(), 
        		type + REDIS_NAMESPACE_SEPARATOR + queueName);
    }
    
    /**
     * 清除此队列与所有type的映射关系
     * @param queueName
     */
    public void clearTypeAndQueueMapping(final String queueName) {
        Set<String> values = zSetOperations.range(slothNameManager.getType2QueueKey(), 0, Long.MAX_VALUE);
        if(CollectionUtils.isEmpty(values)){
            return;
        }
        
        redisLockTemplate.execute(slothNameManager.getType2QueueKey(), new LockCallbackHandler<Boolean>() {

            @Override
            public Boolean doInLock() throws SlothException {
                for(String valueItem : values){
                    if(valueItem.endsWith(queueName)){
                        zSetOperations.remove(slothNameManager.getType2QueueKey(), valueItem);
                    }
                }
                
                return null;
            }
        });
    }
    
    /**
     * 记录消息日志
     * @param msg
     * @param queueName
     * @param dealStatus
     */
    public void logMsg(final SlothMessage msg, final String queueName, final int dealStatus) {
        final String value = msg.getId() + REDIS_NAMESPACE_SEPARATOR + queueName;
        
        final Map<String, Object> dealResultMap = new HashMap<>();
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS, dealStatus);
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME, System.currentTimeMillis());
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG, "");
        
        /*zSetOperations.getOperations().execute(new SessionCallback<Object>() {

            @SuppressWarnings({ "rawtypes", "unchecked" })
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                do {
                    operations.multi();
                    //记录消息体
                    operations.opsForHash().put(slothNameManager.getMsgLogKey(), msg.getId(), msg.toJsonObject());
                    //存储处理状态对应的“消息MsgId:处理队列”
                    operations.opsForSet().add(slothNameManager.getDealStatus2MsgIdAndQueueKey(dealStatus), value);
                    //存储消息处理结果
                    operations.opsForHash().putAll(slothNameManager.getDealResultKey(msg.getId(), queueName), dealResultMap);
                    //建立消息与创建时间索引（zset）
                    operations.opsForZSet().add(slothNameManager.getMsg2CreatedTimeKey(), msg.getId(), msg.getTimeStamp());
                } while(CollectionUtils.isEmpty(operations.exec()));
                
                return null;
            }
        });*/
        
        List<String> needLockKeys = new ArrayList<>();
        needLockKeys.add(slothNameManager.getMsgLogKey() + REDIS_NAMESPACE_SEPARATOR + msg.getId());
        needLockKeys.add(slothNameManager.getDealStatus2MsgIdAndQueueKey(dealStatus));
        needLockKeys.add(slothNameManager.getDealResultKey(msg.getId(), queueName));
        needLockKeys.add(slothNameManager.getMsg2CreatedTimeKey() + REDIS_NAMESPACE_SEPARATOR + msg.getId());
        
        redisLockTemplate.execute(needLockKeys, new LockCallbackHandler<Boolean>() {

            @Override
            public Boolean doInLock() throws SlothException {
                //记录消息体
                hashOperations.put(slothNameManager.getMsgLogKey(), msg.getId(), msg.toJsonObject());
                //存储处理状态对应的“消息MsgId:处理队列”
                setOperations.add(slothNameManager.getDealStatus2MsgIdAndQueueKey(dealStatus), value);
                //存储消息处理结果
                objectHashOperations.putAll(slothNameManager.getDealResultKey(msg.getId(), queueName), dealResultMap);
                //建立消息与创建时间索引（zset）
                zSetOperations.add(slothNameManager.getMsg2CreatedTimeKey(), msg.getId(), msg.getTimeStamp());
                
                return null;
            }
        });
    }
    
    /**
     * 更新消息处理状态
     * @param msgId
     * @param queueName
     * @param dealStatus
     */
    public void updateDealStatusOfMsgForLog(final String msgId, final String queueName, final int dealStatus) {
        updateDealStatusOfMsgForLog(msgId, queueName, dealStatus, null);
    }
    
    /**
     * 更新消息处理状态以及错误原因
     * @param msgId
     * @param queueName
     * @param dealStatus
     * @param errorMsg
     */
    public void updateDealStatusOfMsgForLog(final String msgId, final String queueName, final int dealStatus, final String errorMsg) {
        final String value = msgId + REDIS_NAMESPACE_SEPARATOR + queueName;
        final String dealResultHashKey = slothNameManager.getDealResultKey(msgId, queueName);
        final String dealStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(dealStatus);
        
        final Map<String, Object> dealResultMap = new HashMap<>();
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS, dealStatus);
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME, System.currentTimeMillis());
        dealResultMap.put(SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG, StringUtils.isNotEmpty(errorMsg) ? errorMsg : "");
        
        /*final List<String> keys = new ArrayList<>();
        keys.add(dealStatusSetKey);
        keys.add(dealResultHashKey);
        
        zSetOperations.getOperations().execute(new SessionCallback<Object>() {

            @SuppressWarnings({ "rawtypes", "unchecked" })
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                do {
                    operations.watch(keys);
                    
                    //获取旧的处理状态
                    Object originDealStatus = operations.opsForHash().get(dealResultHashKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS);
                    
                    operations.multi();
                    
                    //更新处理状态对应的“消息MsgId:处理队列”
                    if(originDealStatus != null) {
                    	operations.opsForSet().remove(slothNameManager.getDealStatus2MsgIdAndQueueKey((Integer) originDealStatus), value);
                    }
                    operations.opsForSet().add(dealStatusSetKey, value);
                    //更新消息处理结果
                    operations.opsForHash().putAll(dealResultHashKey, dealResultMap);
                } while(CollectionUtils.isEmpty(operations.exec()));
                
                return null;
            }
        });*/
        
        //获取旧的处理状态
        Object originDealStatus = objectHashOperations.get(dealResultHashKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS);
        
        List<String> needLockKeys = new ArrayList<>();
        if(originDealStatus != null) {
            needLockKeys.add(slothNameManager.getDealStatus2MsgIdAndQueueKey((Integer) originDealStatus));
        }
        needLockKeys.add(dealResultHashKey);
        needLockKeys.add(dealStatusSetKey);
        
        redisLockTemplate.execute(needLockKeys, new LockCallbackHandler<Boolean>() {

            @Override
            public Boolean doInLock() throws SlothException {
                //更新处理状态对应的“消息MsgId:处理队列”
                if(originDealStatus != null) {
                    setOperations.remove(slothNameManager.getDealStatus2MsgIdAndQueueKey((Integer) originDealStatus), value);
                }
                setOperations.add(dealStatusSetKey, value);
                //更新消息处理结果
                objectHashOperations.putAll(dealResultHashKey, dealResultMap);
                
                return null;
            }
        });
    }
    
    /**
     * 清除早期消息日志
     * @param earlyTempStamp 日志最早处理时间戳
     */
    public void clearEarlyMsgLog(final long earlyTempStamp) {
        Long logSize = hashOperations.size(slothNameManager.getMsgLogKey());
        if(logSize == null){
            return;
        }
        
        String newStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_NEW);
        String executingStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXECUTING);
        String successStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_SUCCESS);
        String failStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_FAIL);
        
        /*zSetOperations.getOperations().execute(new SessionCallback<Object>() {

            @SuppressWarnings({ "rawtypes", "unchecked" })
            @Override
            public Object execute(RedisOperations operations) throws DataAccessException {
                do {
                    //获取需要删除的数据
                    Set<String> values = operations.opsForZSet().rangeByScore(slothNameManager.getMsg2CreatedTimeKey(), 0, earlyTempStamp);
                    if(CollectionUtils.isEmpty(values)){
                        operations.unwatch();
                        return null;
                    }
                    
                    List<String> dealResultKeys = new ArrayList<>();
                    for(String item : values){
                    	Set<String> keys = operations.keys(slothNameManager.getDealResultKeyPrefix(item) + "*");
                    	if(CollectionUtils.isEmpty(keys)) {
                    		continue;
                    	}
                    	dealResultKeys.addAll(keys);
                    }
                    
                    operations.multi();
                    
                    //删除早期数据
                    for(String item : values){
                    	//清除消息体
                        operations.opsForHash().delete(slothNameManager.getMsgLogKey(), item);
                        //清除消息与创建时间索引
                        operations.opsForZSet().remove(slothNameManager.getMsg2CreatedTimeKey(), item);
                    }
                    for(String item : dealResultKeys) {
                    	//清除消息处理结果
                    	operations.delete(item);
                    	//清除处理状态对应的“消息MsgId:处理队列”对应关系
                    	String msgIdAndQueue = item.substring(slothNameManager.getDealResultKeyPrefix().length());
                    	operations.opsForSet().remove(newStatusSetKey, msgIdAndQueue);
                    	operations.opsForSet().remove(executingStatusSetKey, msgIdAndQueue);
                    	operations.opsForSet().remove(successStatusSetKey, msgIdAndQueue);
                    	operations.opsForSet().remove(failStatusSetKey, msgIdAndQueue);
                    }
                } while(CollectionUtils.isEmpty(operations.exec()));
                
                return null;
            }
        });*/
        
        //获取需要删除的数据
        Set<String> values = zSetOperations.rangeByScore(slothNameManager.getMsg2CreatedTimeKey(), 0, earlyTempStamp);
        if(CollectionUtils.isEmpty(values)){
            return;
        }
        
        List<String> needLockKeys = new ArrayList<>();
        for(String msgIdItem : values){
            needLockKeys.clear();
            
            needLockKeys.add(newStatusSetKey);
            needLockKeys.add(executingStatusSetKey);
            needLockKeys.add(successStatusSetKey);
            needLockKeys.add(failStatusSetKey);
            needLockKeys.add(slothNameManager.getMsgLogKey() + REDIS_NAMESPACE_SEPARATOR + msgIdItem);
            needLockKeys.add(slothNameManager.getMsg2CreatedTimeKey() + REDIS_NAMESPACE_SEPARATOR + msgIdItem);
            Set<String> dealResultKeys = valueOperations.getOperations().keys(slothNameManager.getDealResultKeyPrefix(msgIdItem) + "*");
            if(CollectionUtils.isNotEmpty(dealResultKeys)) {
                needLockKeys.addAll(dealResultKeys);
            }
            
            redisLockTemplate.execute(needLockKeys, new LockCallbackHandler<Boolean>() {

                @Override
                public Boolean doInLock() throws SlothException {
                    //清除消息体
                    objectHashOperations.delete(slothNameManager.getMsgLogKey(), msgIdItem);
                    //清除消息与创建时间索引
                    zSetOperations.remove(slothNameManager.getMsg2CreatedTimeKey(), msgIdItem);
                    
                    //清楚消息处理结果数据和消息处理的状态数据
                    if(CollectionUtils.isNotEmpty(dealResultKeys)) {
                        for(String dealResultKeyItem : dealResultKeys){
                            //清除消息处理结果
                            hashOperations.getOperations().delete(dealResultKeyItem);
                            //清除处理状态对应的“消息MsgId:处理队列”对应关系
                            String msgIdAndQueue = dealResultKeyItem.substring(slothNameManager.getDealResultKeyPrefix().length());
                            setOperations.remove(newStatusSetKey, msgIdAndQueue);
                            setOperations.remove(executingStatusSetKey, msgIdAndQueue);
                            setOperations.remove(successStatusSetKey, msgIdAndQueue);
                            setOperations.remove(failStatusSetKey, msgIdAndQueue);
                        }
                    }
                    
                    return null;
                }
            }); 
        } 
    }
    
    /**
     * 获取消息处理日志，按处理状态排序
     */
    public List<SlothMessage> listMsgLogOrderByDealStatus(SlothOrder order, long offset, long count) {
        List<SlothMessage> result = new ArrayList<>();
        
        String newStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_NEW);
        String executingStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXECUTING);
        String successStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_SUCCESS);
        String failStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_FAIL);
        String tempDestKey = slothNameManager.getTempKey();
        List<String> otherKeys = new ArrayList<>();
        otherKeys.add(executingStatusSetKey);
        otherKeys.add(successStatusSetKey);
        otherKeys.add(failStatusSetKey);
        
        setOperations.unionAndStore(newStatusSetKey, otherKeys, tempDestKey);
        setOperations.getOperations().expire(tempDestKey, 10L, TimeUnit.SECONDS);
        
        SortQueryBuilder<String> sortQueryBuilder = SortQueryBuilder.sort(tempDestKey);
        sortQueryBuilder.by(slothNameManager.getDealResultKeyPrefix() + "*->" + SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS);
        if(SlothOrder.ASC.equals(order)){
            sortQueryBuilder.order(Order.ASC);
        } else {
            sortQueryBuilder.order(Order.DESC);
        }
        sortQueryBuilder.limit(offset, count);
        
        List<String> sortResult = setOperations.getOperations().sort(sortQueryBuilder.build());
        
        if(CollectionUtils.isEmpty(sortResult)){
            return result;
        }
        
        String[] tempArr = null;
        for(String item : sortResult){
            tempArr = item.split(REDIS_NAMESPACE_SEPARATOR);
            String id = tempArr[0];
            String idValuePrefix = id + REDIS_NAMESPACE_SEPARATOR;
            String queueName = item.substring(idValuePrefix.length());
            
            //获取消息体
            JSONObject msgJsonObj = hashOperations.get(slothNameManager.getMsgLogKey(), id);
            if(msgJsonObj == null){
                continue;
            }
            
            SlothMessage msg = JSONObject.toJavaObject(msgJsonObj, SlothMessage.class);
            
            msg.setQueueName(queueName);
            msg.setData(null);
            
            //获取处理结果
            String dealResultKey = slothNameManager.getDealResultKey(item);
            msg.setUpdatedTime((Long) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME));
            msg.setDealStatus((Integer) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS));
            msg.setErrorMsg((String) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG));
            
            result.add(msg);
        }
        
        return result;
    }
    
    /**
     * 根据消息ID获取消息处理相关日志信息
     * @param id
     * @return
     */
    public SlothMessage getMsgLogDtlById(String id) {
        //获取消息体
        JSONObject msgJsonObj = hashOperations.get(slothNameManager.getMsgLogKey(), id);
        if(msgJsonObj == null){
            return null;
        }
        
        SlothMessage msg = JSONObject.toJavaObject(msgJsonObj, SlothMessage.class);
        
        String patternPrefix = slothNameManager.getDealResultKeyPrefix(id);
        Set<String> dealResultHashKeys = hashOperations.getOperations().keys(patternPrefix + "*");
        if(CollectionUtils.isEmpty(dealResultHashKeys)){
            return msg;
        }
        
        for(String item : dealResultHashKeys){
            String queueName = item.substring(patternPrefix.length());
            
            Integer status = (Integer) objectHashOperations.get(item, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS);
            Long updatedTime = (Long) objectHashOperations.get(item, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME);
            String errorMsg = (String) objectHashOperations.get(item, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG);
            
            msg.addDealCondition(queueName, status, errorMsg, updatedTime);
        }
        
        return msg;
    }
    
    public SlothMessage getMsgById(String id) {
    	JSONObject msgJsonObj = hashOperations.get(slothNameManager.getMsgLogKey(), id);
        if(msgJsonObj == null){
            return null;
        }
        
        return JSONObject.toJavaObject(msgJsonObj, SlothMessage.class);
    }
    
    /**
     * 获取消息处理日志，按处理时间排序
     */
    public List<SlothMessage> listMsgLogOrderByUpdatedTime(SlothOrder order, long offset, long count) {
        List<SlothMessage> result = new ArrayList<>();
        
        String newStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_NEW);
        String executingStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXECUTING);
        String successStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_SUCCESS);
        String failStatusSetKey = slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_FAIL);
        String tempDestKey = slothNameManager.getTempKey();
        List<String> otherKeys = new ArrayList<>();
        otherKeys.add(executingStatusSetKey);
        otherKeys.add(successStatusSetKey);
        otherKeys.add(failStatusSetKey);
        
        /*setOperations.unionAndStore(newStatusSetKey, otherKeys, tempDestKey);
        setOperations.getOperations().expire(tempDestKey, 10L, TimeUnit.SECONDS);
        
        SortQueryBuilder<String> sortQueryBuilder = SortQueryBuilder.sort(tempDestKey);
        sortQueryBuilder.by(slothNameManager.getDealResultKeyPrefix() + "*->" + SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME);
        if(SlothOrder.ASC.equals(order)){
            sortQueryBuilder.order(Order.ASC);
        } else {
            sortQueryBuilder.order(Order.DESC);
        }
        sortQueryBuilder.limit(offset, count);
        
        List<String> sortResult = setOperations.getOperations().sort(sortQueryBuilder.build());*/
        
        Set<String> dealResultKeyPostfixs = setOperations.union(newStatusSetKey, otherKeys);
        if(CollectionUtils.isEmpty(dealResultKeyPostfixs)){
            return result;
        }
        
        Set<TypedTuple<String>> tuples = new HashSet<>();
        TypedTuple<String> tempTuple = null;
        String tempDealResultKey = null;
        for(String item : dealResultKeyPostfixs){
            tempDealResultKey = slothNameManager.getDealResultKeyPrefix() + item;
            tempTuple = new DefaultTypedTuple<>(item, ((Long) objectHashOperations.get(tempDealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME)).doubleValue());
            tuples.add(tempTuple);
        }
        zSetOperations.add(tempDestKey, tuples);
        zSetOperations.getOperations().expire(tempDestKey, 10L, TimeUnit.SECONDS);
        
        Set<String> sortResult = null;
        if(SlothOrder.ASC.equals(order)){
            sortResult = zSetOperations.range(tempDestKey, offset, offset + count);
        } else {
            sortResult = zSetOperations.reverseRange(tempDestKey, offset, offset + count);
        }
        
        if(CollectionUtils.isEmpty(sortResult)){
            return result;
        }
        
        String[] tempArr = null;
        for(String item : sortResult){
            tempArr = item.split(REDIS_NAMESPACE_SEPARATOR);
            String id = tempArr[0];
            String idValuePrefix = id + REDIS_NAMESPACE_SEPARATOR;
            String queueName = item.substring(idValuePrefix.length());
            
            //获取消息体
            JSONObject msgJsonObj = hashOperations.get(slothNameManager.getMsgLogKey(), id);
            if(msgJsonObj == null){
                continue;
            }
            
            SlothMessage msg = JSONObject.toJavaObject(msgJsonObj, SlothMessage.class);
            
            msg.setQueueName(queueName);
            msg.setData(null);
            
            //获取处理结果
            String dealResultKey = slothNameManager.getDealResultKey(item);
            msg.setUpdatedTime((Long) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME));
            msg.setDealStatus((Integer) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_STATUS));
            msg.setErrorMsg((String) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG));
            
            result.add(msg);
        }
        
        return result;
    }
    
    /**
     * 获取消息处理日志（分页获取）
     * @param page
     * @return
     */
    public List<SlothMessage> listMsgLog(SlothPage page) {
        if("updatedTime".equals(page.getOrderByField())){
            return listMsgLogOrderByUpdatedTime(page.getOrder(), page.getOffset(), page.getCount());
        } else if("dealStatus".equals(page.getOrderByField())){
            return listMsgLogOrderByDealStatus(page.getOrder(), page.getOffset(), page.getCount());
        }
        
        return null;
    }
    
    public List<SlothMessage> listMsgLogByStatus(SlothPage page, int dealStatus) {
        List<SlothMessage> result = new ArrayList<>();
        
        SortQueryBuilder<String> sortQueryBuilder = SortQueryBuilder.sort(slothNameManager.getDealStatus2MsgIdAndQueueKey(dealStatus));
        if(SlothOrder.ASC.equals(page.getOrder())){
        	sortQueryBuilder.order(Order.ASC);
        } else {
        	sortQueryBuilder.order(Order.DESC);
        }
        sortQueryBuilder.alphabetical(true);
        sortQueryBuilder.limit(page.getOffset(), page.getCount());
        
        List<String> msgIdAndQueues = setOperations.getOperations().sort(sortQueryBuilder.build());
        if(CollectionUtils.isEmpty(msgIdAndQueues)){
            return result;
        }
        
        String[] tempArr = null;
        for(String item : msgIdAndQueues){
            tempArr = item.split(REDIS_NAMESPACE_SEPARATOR);
            String id = tempArr[0];
            String idValuePrefix = id + REDIS_NAMESPACE_SEPARATOR;
            String queueName = item.substring(idValuePrefix.length());
            
            //获取消息体
            JSONObject msgJsonObj = hashOperations.get(slothNameManager.getMsgLogKey(), id);
            if(msgJsonObj == null){
                continue;
            }
            
            SlothMessage msg = JSONObject.toJavaObject(msgJsonObj, SlothMessage.class);
            
            msg.setDealStatus(dealStatus);
            msg.setQueueName(queueName);
            msg.setData(null);
            
            //获取处理结果
            String dealResultKey = slothNameManager.getDealResultKey(item);
            msg.setUpdatedTime((Long) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_UPDATED_TIME));
            msg.setErrorMsg((String) objectHashOperations.get(dealResultKey, SlothConstants.LOG_MSG_DEAL_RESULT_HASH_FIELD_ERROR_MSG));
            
            result.add(msg);
        }
        
        return result;
    }
    
    /**
     * 统计当前消息各个处理状态数量值
     * @return
     */
    public JSONObject countMsgStatus() {
        Long newCount = setOperations.size(slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_NEW));
        Long executingCount = setOperations.size(slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXECUTING));
        Long successCount = setOperations.size(slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_SUCCESS));
        Long failCount = setOperations.size(slothNameManager.getDealStatus2MsgIdAndQueueKey(SlothConstants.STATUS_EXEC_FAIL));
        
        JSONObject result = new JSONObject();
        result.put("new", newCount == null ? 0L : newCount);
        result.put("executing", executingCount == null ? 0L : executingCount);
        result.put("success", successCount == null ? 0L : successCount);
        result.put("fail", failCount == null ? 0L : failCount);
        
        return result;
    }
}
